package task5

import java.util.Date

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}

import scala.util.Random

case class RegisterMsg(name: String)
case class RegisterSuccessMsg(msg: String = "success")
case class HeartBeatMsg(name: String, msg: String = "alive")

class MasterActor extends Actor{
  private val workerLst = collection.mutable.HashMap[String, Long]()

  override def preStart(): Unit = {
    // 定时检测map中的worker，剔除超时worker
    import context.dispatcher
    import scala.concurrent.duration._
    context.system.scheduler.schedule(0 seconds, 5 seconds){
      val current: Long = new Date().getTime
      val overtimeWorkers = workerLst.filter(current - _._2 > 5000L).keys
      println(s"超时slavers : ${overtimeWorkers.toString()}")
      if(overtimeWorkers.nonEmpty){
        workerLst --= overtimeWorkers
      }
      println(s"存活slavers : ${workerLst.toString}")
    }
  }

  override def receive: Receive = {
    case RegisterMsg(name) => {
      // worker在master上注册
      println(s"${name} 注册成功!")
      // 将注册信息写入map中
//      workerLst += regMsg.name -> new Date().getTime
      workerLst.put(name, new Date().getTime)
      sender ! RegisterSuccessMsg("success")
    }
    case HeartBeatMsg(name, msg) => {
      println(s"$name 发送心跳")
      // 更新map中的最后心跳时间
      workerLst.put(name, new Date().getTime)
    }
  }
}

class SlaverActor extends Actor{
  private var masterActorRef: ActorSelection = _
  private var name: String = _

  override def preStart(): Unit = {
    // 生成master actor 地址
    masterActorRef = context.actorSelection("akka://sparkFactory/user/master")
    // 从path中提出name
    name = self.path.toString.split("/").last
    // 向master注册
    masterActorRef ! RegisterMsg(name)
  }

  override def receive: Receive = {
    case RegisterSuccessMsg(_) => {
      // 定时发送hearbeat
      import context.dispatcher
      import scala.concurrent.duration._
      context.system.scheduler.schedule(0 seconds, 1 seconds){
        // 加入一个随机数，指定每次发送心跳后睡眠时间，用于模拟slaver宕机
        val sleepTime = Random.nextInt(10) * 1000
        masterActorRef ! HeartBeatMsg(name)
        Thread.sleep(sleepTime)
      }
    }
  }
}

object ImitateSparkDemo {
  def main(args: Array[String]): Unit = {
    val sparkFactory = ActorSystem("sparkFactory")
    val master = sparkFactory.actorOf(Props[MasterActor], "master")
    val slaver1 = sparkFactory.actorOf(Props[SlaverActor], "slaver1")
    val slaver2 = sparkFactory.actorOf(Props[SlaverActor], "slaver2")
  }
}
